特定ポートを許可したセキュリティグループがNLB/ALBにアタッチされたことを検知するAWS Config カスタム Lambda ルールを作成してみた

AWSTemplateFormatVersion: '2010-09-09'
Description: 'Custom AWS Config rule to monitor NLB/ALB security groups for RDP/SSH access'

Parameters:
  ExcludedLoadBalancers:
    Type: String
    Description: 'Comma-separated list of load balancer names to exclude from the check'
    Default: ''

Resources:
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSConfigRulesExecutionRole
      Policies:
        - PolicyName: LambdaExecutionPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - ec2:DescribeSecurityGroups
                  - elasticloadbalancing:DescribeLoadBalancers
                Resource: '*'
              - Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: arn:aws:logs:*:*:*

  ConfigLambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      Handler: index.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Environment:
        Variables:
          EXCLUDED_LOAD_BALANCERS: !Ref ExcludedLoadBalancers
      Code:
        ZipFile: |
          import os
          import boto3
          import json

          def evaluate_compliance(configuration_item, rule_parameters):
              if configuration_item['resourceType'] not in ['AWS::ElasticLoadBalancingV2::LoadBalancer']:
                  return 'NOT_APPLICABLE'

              # Check that configuration_item['configuration'] is not None
              if not configuration_item.get('configuration'):
                  print("Error: configuration is None or not present in configuration_item")
                  return 'NOT_APPLICABLE'

              # Check the existence of loadBalancerArn
              load_balancer_arn = configuration_item['configuration'].get('loadBalancerArn')
              if not load_balancer_arn:
                  print("Error: loadBalancerArn is not present in configuration")
                  return 'NOT_APPLICABLE'

              elb_client = boto3.client('elbv2')
              ec2_client = boto3.client('ec2')

              try:
                  load_balancer = elb_client.describe_load_balancers(LoadBalancerArns=[load_balancer_arn])['LoadBalancers'][0]
              except Exception as e:
                  print(f"Error describing load balancer: {str(e)}")
                  return 'NOT_APPLICABLE'

              # Check if the load balancer is in the exclusion list
              excluded_load_balancers = os.environ.get('EXCLUDED_LOAD_BALANCERS', '').split(',')
              if load_balancer['LoadBalancerName'] in excluded_load_balancers:
                  return 'COMPLIANT'

              security_groups = load_balancer['SecurityGroups']

              for sg_id in security_groups:
                  sg = ec2_client.describe_security_groups(GroupIds=[sg_id])['SecurityGroups'][0]
                  for rule in sg['IpPermissions']:
                      if rule.get('FromPort') in [22, 3389] or rule.get('ToPort') in [22, 3389]:
                          return 'NON_COMPLIANT'

              return 'COMPLIANT'

          def lambda_handler(event, context):
              try:
                invoking_event = json.loads(event['invokingEvent'])
                rule_parameters = json.loads(event['ruleParameters'])

                configuration_item = invoking_event['configurationItem']
                result_token = event['resultToken']

                compliance_type = evaluate_compliance(configuration_item, rule_parameters)

                config = boto3.client('config')
                config.put_evaluations(
                    Evaluations=[
                        {
                            'ComplianceResourceType': configuration_item['resourceType'],
                            'ComplianceResourceId': configuration_item['resourceId'],
                            'ComplianceType': compliance_type,
                            'Annotation': 'Load Balancer security group allows RDP/SSH access' if compliance_type == 'NON_COMPLIANT' else 'Load Balancer security group is compliant',
                            'OrderingTimestamp': configuration_item['configurationItemCaptureTime']
                        },
                    ],
                    ResultToken=result_token
                )
              except Exception as e:
                  print(f"Error in lambda_handler: {str(e)}")
                  return {
                      'statusCode': 500,
                      'body': json.dumps(f'Error occurred: {str(e)}')
                  }
      Runtime: python3.11
      Timeout: 300

  ConfigRule:
    Type: AWS::Config::ConfigRule
    Properties:
      ConfigRuleName: check-lb-security-groups
      Description: 'Check if NLB/ALB security groups allow RDP/SSH access'
      Source:
        Owner: CUSTOM_LAMBDA
        SourceIdentifier: !GetAtt ConfigLambdaFunction.Arn
        SourceDetails:
          - EventSource: aws.config
            MessageType: ConfigurationItemChangeNotification
      Scope:
        ComplianceResourceTypes:
          - AWS::ElasticLoadBalancingV2::LoadBalancer
      InputParameters:
        ExcludedLoadBalancers: !Ref ExcludedLoadBalancers

  ConfigRulePermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt ConfigLambdaFunction.Arn
      Action: lambda:InvokeFunction
      Principal: config.amazonaws.com

Outputs:
  ConfigRuleArn:
    Description: 'ARN of the created Config Rule'
    Value: !GetAtt ConfigRule.Arn
  LambdaFunctionArn:
    Description: 'ARN of the Lambda function'
    Value: !GetAtt ConfigLambdaFunction.Arn